// 
// Created By: Xiaofeng.Gu 2021/03/23
// 

#include "cybertron/node/TaskRunnerPoolClient.hpp"
#include "cybertron/core/UtilTime.hpp"
#include "cybertron/network/SocketTcpClient.hpp"
#include "cybertron/network/SocketTcpClientSync.hpp"
#include "cybertron/network/Message.hpp"
#include "cybertron/core/Log.hpp"
#include "cybertron/core/UtilStateHelper.hpp"
CYBERTRON_BEGIN

CybertronCoreAPI TaskRunnerPoolClient::TaskRunnerPoolClient()
{
	zeroMembers();
}

CybertronCoreAPI TaskRunnerPoolClient::~TaskRunnerPoolClient()
{
}

CybertronCoreAPI void TaskRunnerPoolClient::enable(bool enable)
{
	mbEnable = enable;
}
CybertronCoreAPI ESocketStatus TaskRunnerPoolClient::getSocketStatus() {
	return mTcpClient.getStatus();
}
CybertronCoreAPI void TaskRunnerPoolClient::closeSocket() {
	mTcpClient.close();
}
CybertronCoreAPI bool TaskRunnerPoolClient::isEnable()
{
	return mbEnable;
}

CybertronCoreAPI bool TaskRunnerPoolClient::initialize(std::string groupId, Common::EWorkNodeType workType, std::string id,
	std::string description, std::string userData, std::string tag)
{
	if (!mbEnable)
		return false;
	mNodeInfo.groupId = groupId;
	mNodeInfo.workType = workType;
	mNodeInfo.nodeId = id;
	mNodeInfo.description = description;
	mNodeInfo.userData = userData;
	mNodeInfo.tag = tag;
	
	mRunningThread = std::thread(&TaskRunnerPoolClient::run, this);
	return true;
}


CybertronCoreAPI TaskRunnerPool::ETaskRunnerPoolState TaskRunnerPoolClient::getState() {
	return mState;
}

void TaskRunnerPoolClient::run()
{
	
	cybertron::UtilTime::millisecondsUpdate();
	int now = cybertron::UtilTime::millisecondsCurrent();
	while (true)
	{
		try
		{
			if (mTcpClient.getStatus() != ESocketStatus_Connected)
			{
				mTcpClient.close();
				
				bool succeed = mTcpClient.connect(mTaskRunnerPoolServerIp.c_str(), mTaskRunnerPoolServerPort);
				if (succeed &&mTcpClient.getStatus() == ESocketStatus_Connected) {
					logInfo("Connect TaskRunnerPool Server Successful=====> %s:%d", mTaskRunnerPoolServerIp.c_str(), mTaskRunnerPoolServerPort);
					commitRegisterCommand();
					commitReportState();
					continue;
				}

				logError("Connect  TaskRunnerPool Server Failed=====> %s:%d", mTaskRunnerPoolServerIp.c_str(), mTaskRunnerPoolServerPort);
				std::this_thread::sleep_for(std::chrono::seconds(1));
				continue;
			}

			cybertron::UtilTime::millisecondsUpdate();
			int next = cybertron::UtilTime::millisecondsCurrent();
			bool heartbeat = false;
			if (next - now > 10000)
			{
				now = next;
				heartbeat = true;
			}

			runCommands(heartbeat);
			cybertron::Message message;
			
			int ret = mTcpClient.receiveEnhanceEx(message, /*TimeoutMS*/5000);
			if (ret > 0)
			{
				onServerMessage(message);
			}
			else if (ret == 0) {
				//logError("NodeSONClient receive time out=====>");
			}
			else {
				logError("NodeSONClient receive error=====>");
			}
		}
		catch (std::exception e)
		{
			logError("NodeSONClient receive run catch %s",e.what());
			mTcpClient.close();
			continue;
		}
	}
}


void TaskRunnerPoolClient::onServerMessage(const cybertron::Message& msg)
{
	std::uint16_t msgId = msg.parseMsgId();
	switch (msgId)
	{
		case TaskRunnerPool::EMessageType::ETaskRunnerPoolRegisterRet:
		{
			onHandleRegisterRequestMessage(msg);
		}
		break;

		case Common::EMessageType::ESimOneKeepAliveRet:
		{
			onHandleHeartbeatMessage(msg);
		}
		break;

		//case TaskRunnerPool::ESONMessageTypes::ESONQueryResponse:
		//{
		//	onHandleQueryRequestMessage(msg);
		//}
		//break;

		//case TaskRunnerPool::ESONMessageTypes::ESONStatusUpdateNotice:
		//{
		//	onHandleStatusUpdateNoticeMessage(msg);
		//}

		//break;
		
		case TaskRunnerPool::EMessageType::ETaskRunnerPoolReportStateReq:
		{
			onHandleNodeStateReqMessage(msg);
		}
		case TaskRunnerPool::EMessageType::ETaskRunnerPoolReportStateRet:
		{
			onHandleNodeStateRetMessage(msg);
		}

		break;
	}
}

void TaskRunnerPoolClient::sendHeartbeat()
{
	Common::SimOneKeepAlive keepAlive;
	cybertron::UtilTime::millisecondsUpdate();
	keepAlive.set_timestamp(cybertron::UtilTime::millisecondsCurrent());
	cybertron::Message message;
	message.fromProtobuf(Common::EMessageType::ESimOneKeepAlive, keepAlive);
	mTcpClient.send(message);
}

void TaskRunnerPoolClient::onHandleHeartbeatMessage(const cybertron::Message& msg)
{
	Common::SimOneKeepAliveRet keepAlive;

	if (!msg.toProtobuf(keepAlive)) {
		return;
	}
	cybertron::UtilTime::millisecondsUpdate();
	unsigned int timestamp = cybertron::UtilTime::millisecondsCurrent();
	::google::protobuf::int64 responseTimestamp = keepAlive.timestamp();
	::google::protobuf::int64 cost = timestamp - responseTimestamp;
	//logInfo("onHandleHeartbeatMessage ===>(%lld)", cost);

}

void TaskRunnerPoolClient::onHandleRegisterRequestMessage(const cybertron::Message& msg)
{
	TaskRunnerPool::TaskRunnerPoolRegisterRet data;
	if (!msg.toProtobuf(data))
	{
		return;
	}

	logInfo("onHandleRegisterRequestMessage ===>(RegisterRequestMessage)");
}

void TaskRunnerPoolClient::onHandleNodeStateReqMessage(const cybertron::Message& msg)
{
	TaskRunnerPool::TaskRunnerPoolReportStateReq message;
	if (!msg.toProtobuf(message)) {
		return;
	}
	mState = message.state();

	string workType = UtilStateHelper::getWorkNodeTypeName(mNodeInfo.workType);
	string key = GetKey(mNodeInfo.groupId, mNodeInfo.nodeId, workType);
	string state = UtilStateHelper::getTaskRunnerPoolStateName(mState);
	logInfo("Recv State=========>key:%s state:%s ", key.c_str(), state.c_str());
}

void TaskRunnerPoolClient::onHandleNodeStateRetMessage(const cybertron::Message& msg)
{
	TaskRunnerPool::TaskRunnerPoolReportStateRet message;
	if (!msg.toProtobuf(message)) {
		return;
	}
	//mState = message.state();
	//logInfo("onHandleNodeStateRetMessage ===>(mState:%d)", mState);
}

CybertronCoreAPI bool TaskRunnerPoolClient::setServerInfo(const std::string& serverAddr, int port) {
	mTaskRunnerPoolServerIp = serverAddr;
	mTaskRunnerPoolServerPort = port;
	return true;
}


void TaskRunnerPoolClient::pushCommand(const std::string& command)
{
	mCommandQueue.push_back(command);
}

int TaskRunnerPoolClient::runCommands(bool hasHeartbeat)
{
	int count = 0;
	if (hasHeartbeat)
	{
		sendHeartbeat();
		count++;
		return count;
	}

	std::string command;
	while (mCommandQueue.pop_swap_front(command))
	{
		bool ret = mTcpClient.send(command);
		if (!ret) {
			logError("NodeSONClient Send Error");
			break;
		}
		//logInfo("NodeSONClient ====>runCommands send data");
		count++;
		//break;
	}
	return count;
}

void TaskRunnerPoolClient::updateTaskStateToServer(TaskRunnerPool::ETaskRunnerPoolState state)
{
	if (!mbEnable)
		return ;
	mState = state;
	commitReportState();
}


void TaskRunnerPoolClient::commitRegisterCommand()
{
	TaskRunnerPool::TaskRunnerPoolRegisterReq req;
	req.set_nodeid(mNodeInfo.nodeId);
	req.set_worktype(mNodeInfo.workType);
	req.set_groupid(mNodeInfo.groupId);
	req.set_description(mNodeInfo.description);
	req.set_userdata(mNodeInfo.userData);
	req.set_tag(mNodeInfo.tag);

	cybertron::Message message;
	message.fromProtobuf(TaskRunnerPool::ETaskRunnerPoolRegisterReq, req);
	std::string registerRequestData((char*)message.getData(), message.getLength());
	pushCommand(registerRequestData);

	string workType = UtilStateHelper::getWorkNodeTypeName(mNodeInfo.workType);
	string key = GetKey(mNodeInfo.groupId, mNodeInfo.nodeId, workType);

	logInfo("TaskRunnerPoolClient=========>register key:%s", key.c_str());

}
string TaskRunnerPoolClient::GetKey(string groupId, string id, string type) {
	string key = groupId + "_" + id + "_" + type;
	return key;
}

void TaskRunnerPoolClient::commitReportState() {

	TaskRunnerPool::TaskRunnerPoolReportStateReq req;
	req.set_state(mState);
	req.set_nodeid(mNodeInfo.nodeId);
	req.set_worktype(mNodeInfo.workType);
	req.set_groupid(mNodeInfo.groupId);

	cybertron::Message message;
	message.fromProtobuf(TaskRunnerPool::ETaskRunnerPoolReportStateReq, req);
	std::string data((char*)message.getData(), message.getLength());
	pushCommand(data);
	
	//getTaskRunnerPoolStateName()
	string workType =  UtilStateHelper::getWorkNodeTypeName(mNodeInfo.workType);
	string key = GetKey(mNodeInfo.groupId, mNodeInfo.nodeId, workType);
	string state = UtilStateHelper::getTaskRunnerPoolStateName(mState);
	logInfo("Sync State=========>state:%s key:%s", state.c_str(), key.c_str());
}
void TaskRunnerPoolClient::zeroMembers()
{
	mCommandQueue.clear();
	mState = TaskRunnerPool::ETaskRunnerPoolState::ETaskRunnerPoolStateReadying;
	string state = UtilStateHelper::getTaskRunnerPoolStateName(mState);
	logInfo("Init State=========>state:%s", state.c_str());
	
}


CYBERTRON_END
